package com.burges.net.tableAPIAndSQL.table

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{TableEnvironment, Types}
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/14 
  * 描述      TableSchema使用代码示例
  */
object TableSchema {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		val tableEnvironment = TableEnvironment.getTableEnvironment(environment)
		tableEnvironment
				.connect(new FileSystem().path(""))
				.withFormat(
					new Csv()
							.field("field1", Types.STRING)  //根据顺序指定字段名称和类型           必选
							.field("field2", Types.SQL_TIMESTAMP) //根据顺序指定字段名称和类型     必选
							.fieldDelimiter(",")            // 指定列切割符，默认使用逗号             可选
							.lineDelimiter("\n")            // 指定行切割符，默认使用换行符            可选
							.quoteCharacter('"')            // 指定字符串中的单个字符，默认为空        可选
							.commentPrefix("#")             // 指定Comment的前，默认为空             可选
							.ignoreFirstLine()                     // 是否忽略第一行                       可选
							.ignoreParseErrors()                   // 是否忽略解析错误的数据，默认开启        可选
				)
		//通过TableSchema完成映射
//        		.withSchema(
					// 根据字段顺序生成与数据源中对应的字段映射
//					new Schema()
//        			    .field("id", Types.INT)        //指定第一个字段的名称和类型
//						.field("name", Types.STRING)   //指定第二个字段的名称和类型
//						.field("value", Types.BOOLEAN) //指定第三个字段的名称和类型
//
					//除了指定名称和类型外，也可以通过使用proctime和rowtime等方法获取外部数据中的时间属性
//					new Schema()
//        			    .field("Field1", Types.SQL_TIMESTAMP)
//        			    .proctime()            // 获取ProcessTime属性
//        			    .field("Field2", Types.SQL_TIMESTAMP)
//        			    .rowtime(
							/**
							  * 基于EventTime时间概念处理数据，需要生成事件时间RowTime信息，以及watermark的生成逻辑
							  */
							//可以根据字段名称从输入数据中提取
//							new RowTime().timestampsFromFile("ts_field")
							//或者从底层DataStreamAPI中转换而来，数据源需要支持分配时间戳
//							new RowTime().timestampsFromSource()
							//或者通过自定义实现timestampsFromExactor抽取RowTime
//							new RowTime().timestampsFromExtractor(...)

							//指定WaterMark策略
							//延时两秒生成waterMark
//							new RowTime().watermarksPeriodicBounded(2000)
							//和rowtime最大时间保持一致
//							new RowTime().watermarksPeriodicAscending()
							//使用底层DataStreamAPI内键的watermark
//							new RowTime().watermarksFromSource()
						// )             // 获取EventTime属性  需要定义时间字段以及Warkmark生成逻辑
//						.field("Field3", Types.BOOLEAN)
//        		        .from("origin_field_name")  //从Input/Output数据指定字段中获取数据     通过使用from方法从数据集中根据名称映射TableSchema字段信息
//				)
	}

}
